from fastapi import APIRouter, Body
from typing import Dict, Any
from intelli_port.data_layer.clients import mysql_execute_write
from intelli_port.data_clients import health_snapshot
from intelli_port.commons.observability import histogram_p95_ms, http_request_latency_ms, provider_error_rate, histogram_p95_ms_by_label, provider_request_latency_ms, counter_sum_by_label, middleware_idempotency_failures_total, db_query_latency_ms, db_errors_total, db_queries_total, db_slow_queries_total, provider_queue_length, provider_kv_cache_hit_rate, provider_gpu_mem_used_mb, gauge_value_by_label
from intelli_port.commons.config import load_settings
from intelli_port.integrations.fids import sync_fids
from intelli_port.robot_scheduler.service import autoschedule_pending_requests
from intelli_port.data_layer.clients import mysql_execute_read, neo4j_exec
from intelli_port.robot_control.executor import create_robot_executor


router = APIRouter()
executor = create_robot_executor()


@router.get("/slo")
async def slo():
    return {
        "response_p95_ms": 300,
        "api_success_rate_target": 0.99,
        "cache_hit_rate_target": 0.85,
        "content_fail_rate_max": 0.02,
    }


@router.get("/dependencies")
async def dependencies():
    h = health_snapshot()
    return {
        "mysql": {"up": h.get("mysql", False)},
        "redis": {"up": h.get("redis", False)},
        "milvus": {"up": h.get("milvus", False)},
        "neo4j": {"up": h.get("neo4j", False)},
    }


@router.get("/summary")
async def summary():
    http_p95 = histogram_p95_ms(http_request_latency_ms)
    err_http = provider_error_rate("qwen_http")
    err_vllm = provider_error_rate("qwen_vllm")
    p95_http = histogram_p95_ms_by_label(provider_request_latency_ms, "name", "qwen_http")
    p95_vllm = histogram_p95_ms_by_label(provider_request_latency_ms, "name", "qwen_vllm")
    idem_fail_redis = counter_sum_by_label(middleware_idempotency_failures_total, "reason", "redis_unavailable")
    db_p95 = histogram_p95_ms(db_query_latency_ms)
    db_errs = counter_sum_by_label(db_errors_total, "name", "mysql")
    db_total = counter_sum_by_label(db_queries_total, "name", "mysql")
    db_err_rate = round((db_errs / db_total) if db_total > 0 else 0.0, 4)
    db_slow = counter_sum_by_label(db_slow_queries_total, "name", "mysql")
    return {
        "http_p95_ms": round(http_p95, 2),
        "providers": {
            "qwen_http_error_rate": err_http,
            "qwen_vllm_error_rate": err_vllm,
            "qwen_http_p95_ms": round(p95_http, 2),
            "qwen_vllm_p95_ms": round(p95_vllm, 2),
            "qwen_http_queue_length": gauge_value_by_label(provider_queue_length, "name", "qwen_http"),
            "qwen_vllm_queue_length": gauge_value_by_label(provider_queue_length, "name", "qwen_vllm"),
            "qwen_hybridflow_queue_length": gauge_value_by_label(provider_queue_length, "name", "qwen_hybridflow"),
            "qwen_http_kv_cache_hit_rate": gauge_value_by_label(provider_kv_cache_hit_rate, "name", "qwen_http"),
            "qwen_vllm_kv_cache_hit_rate": gauge_value_by_label(provider_kv_cache_hit_rate, "name", "qwen_vllm"),
            "qwen_hybridflow_kv_cache_hit_rate": gauge_value_by_label(provider_kv_cache_hit_rate, "name", "qwen_hybridflow"),
            "qwen_http_gpu_mem_used_mb": gauge_value_by_label(provider_gpu_mem_used_mb, "name", "qwen_http"),
            "qwen_vllm_gpu_mem_used_mb": gauge_value_by_label(provider_gpu_mem_used_mb, "name", "qwen_vllm"),
            "qwen_hybridflow_gpu_mem_used_mb": gauge_value_by_label(provider_gpu_mem_used_mb, "name", "qwen_hybridflow"),
        },
        "idempotency_failures": {"redis_unavailable": int(idem_fail_redis)},
        "db": {"mysql_p95_ms": round(db_p95, 2), "mysql_error_rate": db_err_rate, "mysql_slow_queries": int(db_slow)},
    }


@router.get("/errors")
async def errors():
    return {
        "errors": [
            {"code": "IDEMPOTENT", "status": 409, "message": "Duplicate request"},
            {"code": "REDIS_UNAVAILABLE", "status": 503, "message": "Idempotency storage unavailable"},
            {"code": "UNAUTHORIZED", "status": 401, "message": "Signature verification failed"},
            {"code": "BAD_REQUEST", "status": 400, "message": "Invalid request"},
        ]
    }


@router.get("/policy")
async def policy():
    s = load_settings()
    return {
        "error_rate_threshold": float(getattr(s, "qwen_error_rate_threshold", 0.01)),
        "provider_p95_max_ms": int(getattr(s, "provider_p95_max_ms", 1000)),
        "provider_queue_max": float(getattr(s, "provider_queue_max", 1.0)),
        "provider_gpu_mem_max_mb": float(getattr(s, "provider_gpu_mem_max_mb", 16000.0)),
        "provider_kv_min_hit_rate": float(getattr(s, "provider_kv_min_hit_rate", 0.1)),
    }


@router.post("/sync/fids")
async def ops_sync_fids():
    return sync_fids()


@router.post("/scheduler/run")
async def ops_scheduler_run():
    return autoschedule_pending_requests()


@router.post("/robot/callback/status")
async def robot_callback_status(payload: Dict[str, Any] = Body(...)):
    rid = int(payload.get("robot_id") or 0)
    tid = int(payload.get("task_id") or 0)
    status = str(payload.get("status") or "")
    if rid <= 0 or tid <= 0 or not status:
        return {"ok": False}
    try:
        mysql_execute_write(
            "UPDATE robot_tasks SET status=%s, updated_at=NOW() WHERE id=%s AND robot_id=%s",
            (status, tid, rid),
        )
        replan = None
        if status.lower() == "failed":
            try:
                rb = mysql_execute_read("SELECT location FROM robots WHERE id=%s", (rid,), fetch="one").get("rows", [])
                origin = str((rb[0] or "") if rb and rb[0] else "")
                last = mysql_execute_read("SELECT payload FROM robot_tasks WHERE id=%s AND robot_id=%s", (tid, rid), fetch="one").get("rows", [])
                dest = None
                retry_count = 0
                if last and last[0]:
                    import json
                    try:
                        pl = json.loads(last[0]) if isinstance(last[0], (str, bytes)) else (last[0] or {})
                    except Exception:
                        pl = {}
                    dest = pl.get("poi")
                    retry_count = int(pl.get("retry_count") or 0)
                if origin and dest:
                    from intelli_port.langgraph_module.subgraphs_navigate import build_navigate_subgraph
                    g = build_navigate_subgraph()
                    out = g.invoke({"slots": {"origin": origin, "destination": dest}, "session": {"prefer_accessible": True}})
                    replan = out.get("service")
                    try:
                        from intelli_port.commons.audit import write_ops_log, audit_event
                        write_ops_log("WARN", "robot.replan", {"robot_id": rid, "task_id": tid, "origin": origin, "destination": dest, "service": replan})
                        audit_event("replan", "robot", "navigate", {"robot_id": rid, "task_id": tid, "origin": origin, "destination": dest}, result="suggestion")
                    except Exception:
                        pass
                    # 调度层记录重试次数与原因，并按策略自动重试
                    try:
                        import json, asyncio
                        s = load_settings()
                        new_retry = retry_count + 1
                        pay_upd = json.dumps({"poi": dest, "retry_count": new_retry, "last_reason": status}, ensure_ascii=False)
                        mysql_execute_write("UPDATE robot_tasks SET payload=%s WHERE id=%s AND robot_id=%s", (pay_upd, tid, rid))
                        mysql_execute_write("INSERT INTO robot_events(robot_id, task_id, event, detail, created_at) VALUES(%s,%s,%s,%s,NOW())", (rid, tid, "retry", json.dumps({"count": new_retry}, ensure_ascii=False)))
                        if getattr(s, "robot_nav_auto_retry", True) and new_retry <= int(getattr(s, "robot_nav_retry_max", 1)):
                            try:
                                rr = neo4j_exec("MATCH (p:POI {name:$name}) RETURN COALESCE(p.x,0.0) AS x, COALESCE(p.y,0.0) AS y, COALESCE(p.yaw,0.0) AS yaw", {"name": dest}) or []
                                if rr and rr[0]:
                                    target = {"goal_x": float(rr[0].get("x") or 0.0), "goal_y": float(rr[0].get("y") or 0.0), "yaw": float(rr[0].get("yaw") or 0.0)}
                                    backoff_ms = int(getattr(s, "robot_nav_retry_backoff_ms", 0) or 0)
                                    # Enqueue to Redis worker; fallback to immediate navigate if Redis unavailable
                                    try:
                                        from intelli_port.data_clients.redis import get_redis
                                        rds = get_redis()
                                        import json
                                        rds.rpush("robot:retry", json.dumps({"robot_id": rid, "task_id": tid, "target": target, "backoff_ms": backoff_ms}, ensure_ascii=False))
                                    except Exception:
                                        try:
                                            _ = executor.navigate_to(rid, {"type": "navigate", "target": target, "task_id": tid})
                                        except Exception:
                                            pass
                            except Exception:
                                pass
                    except Exception:
                        pass
            except Exception:
                replan = None
        return {"ok": True, "replan": replan}
    except Exception:
        return {"ok": False}


@router.post("/robot/command")
async def robot_command(payload: Dict[str, Any] = Body(...)):
    rid = int(payload.get("robot_id") or 0)
    t = payload.get("type") or "navigate"
    target = payload.get("target") or {}
    if rid <= 0:
        return {"ok": False}
    try:
        poi = target.get("poi")
        if poi:
            rr = neo4j_exec("MATCH (p:POI {name:$name}) RETURN COALESCE(p.x,0.0) AS x, COALESCE(p.y,0.0) AS y, COALESCE(p.yaw,0.0) AS yaw", {"name": poi}) or []
            if rr and rr[0]:
                target = {"goal_x": float(rr[0].get("x") or 0.0), "goal_y": float(rr[0].get("y") or 0.0), "yaw": float(rr[0].get("yaw") or 0.0)}
    except Exception:
        pass
    tid = int(payload.get("task_id") or 0)
    r = executor.navigate_to(rid, {"type": t, "target": target, "task_id": tid})
    try:
        import json
        mysql_execute_write(
            "UPDATE robot_tasks SET payload=%s WHERE id=%s AND robot_id=%s",
            (json.dumps(target, ensure_ascii=False), tid, rid),
        )
    except Exception:
        pass
    return r


@router.get("/robot/status")
async def robot_status(robot_id: int):
    return executor.status(robot_id)


@router.post("/robot/cancel")
async def robot_cancel(payload: Dict[str, Any] = Body(...)):
    rid = int(payload.get("robot_id") or 0)
    tid = int(payload.get("task_id") or 0)
    if rid <= 0 or tid <= 0:
        return {"ok": False}
    return executor.cancel(rid, tid)


@router.get("/robot/events")
async def robot_events(robot_id: int, limit: int = 50):
    try:
        rows = mysql_execute_read(
            "SELECT event, detail, created_at FROM robot_events WHERE robot_id=%s ORDER BY created_at DESC LIMIT %s",
            (robot_id, limit),
            fetch="all",
        ).get("rows", [])
        events = [{"event": x[0], "detail": x[1], "at": x[2].isoformat() if x[2] else None} for x in rows]
        return {"events": events}
    except Exception:
        return {"events": []}


@router.get("/scheduler/plan")
async def ops_scheduler_plan(limit: int = 10):
    s = load_settings()
    reqs = mysql_execute_read(
        "SELECT id, service_type, location FROM service_requests WHERE status='pending' ORDER BY priority DESC, id ASC LIMIT %s",
        (limit,),
        fetch="all",
    ).get("rows", [])
    robots = mysql_execute_read(
        "SELECT id, serial, location, battery, status FROM robots WHERE status='idle' ORDER BY battery DESC",
        fetch="all",
    ).get("rows", [])
    plans = []
    for r in reqs:
        rid = int(r[0])
        rloc = str(r[2] or "")
        cand = []
        for rb in robots:
            rb_id = int(rb[0])
            rb_loc = str(rb[2] or "")
            batt = float(rb[3] or 0.0)
            dist = 0.0
            tmin = 0.0
            crowd = 1.0
            try:
                if rloc and rb_loc:
                    q = (
                        "MATCH (a:POI {name:$src}),(b:POI {name:$dst}) CALL algo.shortestPath.stream(a,b) YIELD nodeId WITH nodeId MATCH (n) WHERE id(n)=nodeId RETURN n.name AS name"
                    )
                    names = [x.get("name") for x in (neo4j_exec(q, {"src": rb_loc, "dst": rloc}) or []) if x.get("name")]
                    if names and len(names) > 1:
                        dsum = 0.0
                        tsum = 0.0
                        csum = 0.0
                        for i in range(len(names) - 1):
                            a = names[i]
                            b = names[i + 1]
                            rr = neo4j_exec(
                                "MATCH (a:POI {name:$a})-[e:PATH]->(b:POI {name:$b}) RETURN COALESCE(e.distance_m,0.0) AS dm, COALESCE(e.time_s,0.0) AS ts, COALESCE(e.crowd_factor,1.0) AS cf",
                                {"a": a, "b": b},
                            ) or []
                            if rr and rr[0]:
                                dsum += float(rr[0].get("dm") or 0.0)
                                tsum += float(rr[0].get("ts") or 0.0)
                                csum += float(rr[0].get("cf") or 1.0)
                        dist = dsum
                        tmin = (tsum / 60.0) if tsum > 0 else 0.0
                        crowd = (csum / max(1, (len(names) - 1))) if csum > 0 else 1.0
            except Exception:
                dist = 0.0
                tmin = 0.0
                crowd = 1.0
            wdist = float(getattr(s, "scheduler_w_dist", 1.0))
            wtime = float(getattr(s, "scheduler_w_time", 1.0))
            wbatt = float(getattr(s, "scheduler_w_batt", 1.0))
            wcrowd = float(getattr(s, "scheduler_w_crowd", 0.5))
            wskill = float(getattr(s, "scheduler_w_skill", 1.0))
            batt_pen = 0.0 if batt >= 30.0 else (30.0 - batt)
            skill = 0.0
            score = wdist * dist + wtime * tmin + wbatt * batt_pen + wcrowd * crowd + wskill * skill
            cand.append({"robot_id": rb_id, "location": rb_loc, "battery": batt, "distance_m": dist, "duration_min": tmin, "crowd": crowd, "score": round(score, 3)})
        cand.sort(key=lambda x: x["score"])
        plans.append({"request_id": rid, "candidates": cand[:3]})
    return {"plans": plans}